package com.sunday.common.reactor.e7_Handling_Errors;

import com.sunday.common.reactor.SubscribeUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * A.5. 处理错误
 */
@Slf4j
public class HandlingErrors {

    /**
     * 我想创建一个错误序列：（Flux|Mono）...error
     * 以替换成功完成的 Flux：.concat(Flux.error(e))
     * 替换成功的单声道发射：.then(Mono.error(e))
     * 如果 onNext 之间间隔时间过长：（Flux|Mono#timeout)
     * 懒惰地：（Flux|Mono#error(Supplier<Throwable>))
     */
    @Test
    public void error() throws InterruptedException {

        log.info("Flux.error 用于创建一个立即发出错误信号的 Flux，通过提供相应的 Throwable 对象，可以指定要引发的特定错误。");
        // Flux.error 用于创建一个立即发出错误信号的 Flux，通过提供相应的 Throwable 对象，可以指定要引发的特定错误。
        SubscribeUtils.subscribe(Flux.error(new ArithmeticException("/ by zero")));

        log.info("Flux.concat(Flux.error(e)) 负责将多个 Flux 连接在一起，按照它们被提供的顺序依次发出它们的元素，非常适合需要串行处理多个数据流的情况。");
        // Flux.concat(Flux.error(e)) 负责将多个 Flux 连接在一起，按照它们被提供的顺序依次发出它们的元素，非常适合需要串行处理多个数据流的情况。
        SubscribeUtils.subscribe(Flux.concat(Flux.just(1, 2, 3), Flux.error(new ArithmeticException("/ by zero"))));

        log.info(".then 用于在成功完成时发出完成信号，或者在错误时发出错误信号。");
        // .then 用于在成功完成时发出完成信号，或者在错误时发出错误信号。
        SubscribeUtils.subscribe(Mono.just("1").delayElement(Duration.ofSeconds(1)).then(Mono.error(new RuntimeException("休眠1s后执行的报错"))));
        TimeUnit.SECONDS.sleep(2);

        log.info("Flux#timeout 用于在指定的时间间隔内，如果没有发出任何元素，则发出一个错误信号。");
        // Flux#timeout 用于在指定的时间间隔内，如果没有发出任何元素，则发出一个错误信号。
        SubscribeUtils.subscribe(Flux.just(1, 2, 3).delayElements(Duration.ofSeconds(1)).timeout(Duration.ofMillis(500)));
        TimeUnit.SECONDS.sleep(2);

        log.info("Flux#error(Supplier<Throwable>) 用于在需要时才创建错误对象。");
        // Flux#error(Supplier<Throwable>) 用于在需要时才创建错误对象。
        SubscribeUtils.subscribe(Flux.just(1, 2, 3).delayElements(Duration.ofSeconds(1)).timeout(Duration.ofMillis(500)).onErrorResume(e -> {
            log.info("onErrorResume 用于在需要时才创建错误对象。");
            return Flux.error(new RuntimeException("用于在需要时才创建错误对象。"));
        }));
        TimeUnit.SECONDS.sleep(2);
    }

    /**
     * 我想要相当于以下内容的 try/catch：
     */

    /**
     * 投掷：（通量|Mono#error)
     */
    @Test
    public void monoError() {
        log.info("Mono#error 用于创建一个立即发出错误信号的 Mono，通过提供相应的 Throwable 对象，可以指定要引发的特定错误。");
        // Mono#error 用于创建一个立即发出错误信号的 Mono，通过提供相应的 Throwable 对象，可以指定要引发的特定错误。
        SubscribeUtils.subscribe(Mono.error(new ArithmeticException("/ by zero")));
    }

    /**
     * 并回退到默认值：（Flux|Mono#onErrorReturn)
     */
    @Test
    public void onErrorReturn() {
        log.info("Flux#onErrorReturn 用于在遇到错误时提供一个备用值。");
        // Flux#onErrorReturn 用于在遇到错误时提供一个备用值。
        SubscribeUtils.subscribe(Flux.just(1, 2, 3).map(i -> 10 / (i - 2)).onErrorReturn(0));
    }

    /**
     * 并吞下错误（即完成）：（Flux|Mono#onErrorComplete)
     */
    @Test
    public void onErrorComplete() {
        log.info("Flux#onErrorComplete 用于在遇到错误时忽略错误信号，而是发出一个完成信号。");
        // Flux#onErrorComplete 用于在遇到错误时忽略错误信号，而是发出一个完成信号。
        SubscribeUtils.subscribe(Flux.just(1, 2, 3).map(i -> 10 / (i - 2)).onErrorComplete());
    }

    /**
     * 并回退到另一个 Flux 或 Mono：（Flux|Mono#onErrorResume)
     */
    @Test
    public void onErrorResume() {
        log.info("Flux#onErrorResume 用于在遇到错误时提供一个备用的 Flux 或 Mono。");
        // Flux#onErrorResume 用于在遇到错误时提供一个备用的 Flux 或 Mono。 用于在需要时才创建错误对象。
        SubscribeUtils.subscribe(Flux.just(1, 2, 3).map(i -> 10 / (i - 2)).onErrorResume(e -> Flux.just(0)));
    }

    /**
     * 以及包装和重新抛出：（Flux|Mono.onErrorMap(t → new RuntimeException(t)))
     */
    @Test
    public void onErrorMap() {
        log.info("Flux#onErrorMap 用于在遇到错误时提供一个备用的 Flux 或 Mono。");
        // Flux#onErrorMap 用于在遇到错误时提供一个备用的 Flux 或 Mono。 用于在需要时才创建错误对象。
        SubscribeUtils.subscribe(Flux.just(1, 2, 3).map(i -> 10 / (i - 2)).onErrorMap(e -> new RuntimeException(e)));
    }

    /**
     * 最后一个块：（Flux|Mono#doFinally)
     */
    @Test
    public void doFinally() {
        log.info("Flux#doFinally 用于在序列完成或终止时执行清理操作。");
        // Flux#doFinally 用于在序列完成或终止时执行清理操作。
        SubscribeUtils.subscribe(Flux.just(1, 2, 3).map(i -> 10 / (i - 2)).doFinally(s -> log.info("doFinally")));
    }

    /**
     * Java 7 中的使用模式：（Flux|Mono）工厂法using
     * 我想要相当于以下内容的 try-with-resources：
     */
    @Test
    public void using() {
        log.info("Flux#using 用于在序列完成或终止时执行清理操作。");
        // Flux#using 用于在序列完成或终止时执行清理操作。
        SubscribeUtils.subscribe(Flux.using(() -> "resource", resource -> Flux.just(resource + "1", resource + "2"), resource -> log.info("清理资源 {}", resource)));
    }

    /**
     * 通过重试...
     */

    /**
     * 重试 (异步)
     * 在错误处理方面，还有另一个感兴趣的运算符，您可能是 很想在上一节所述的情况下使用它。，正如它的名字 指示，允许您重试产生错误的序列。retry
     * https://projectreactor.io/docs/core/release/reference/#_retrying
     * <p>
     * subscribe : [268,tick 0]
     * subscribe : [248,tick 1]
     * subscribe : [246,tick 2]
     * 一个新的开始，从刻度 0 开始。额外的 250 毫秒持续时间为 来自第 4 个分时，导致异常的分时和随后的分时 重试
     * subscribe : [510,tick 0]
     * subscribe : [248,tick 1]
     * subscribe : [248,tick 2]
     * Error : java.lang.RuntimeException: boom
     */
    @Test
    public void retry() throws InterruptedException {
        Flux<String> flux = Flux.interval(Duration.ofMillis(250)).map(input -> {
                    if (input < 3) return "tick " + input;
                    throw new RuntimeException("boom");
                })
                // 	重试产生错误的序列。retry
                .retry(1);
        // elapsed 返回一个 Flux，它发出一个元素，该元素是相对于前一个元素的持续时间。elapsed
        SubscribeUtils.subscribe(flux.elapsed());
        TimeUnit.SECONDS.sleep(5);
    }

    /**
     * 重试周期如下： (同步)
     * 每次发生错误（有可能重试）时，都会向 伴侣，已由您的功能装饰。在这里有一个 鸟瞰迄今为止的所有尝试。提供对错误的访问权限 以及围绕它的元数据。RetrySignalFluxFluxRetrySignal
     * 如果随播对象发出值，则会重试。Flux
     * 如果伴随完成，则吞下错误，重试周期停止， 生成的序列也完成。Flux
     * 如果伴随事件产生错误 （），则重试周期将停止，并且 导致的序列错误。Fluxee
     */
    @Test
    public void retryWhen() {
        Flux<String> flux = Flux
                // 这会不断产生错误，需要重试。
                .<String>error(new IllegalArgumentException())
                // doOnError在重试之前，我们可以记录并查看所有失败。
                .doOnError(e -> log.info("error : {}", e.toString()))
                // Retry.from: Retry.from 方法是一个静态工厂方法，它接受一个 Retry 对象作为参数，该对象定义了重试逻辑。这个方法允许您以自定义的方式对错误进行重试。
                // Retry 对象可以包含要重试的次数、重试的条件、重试时要应用的延迟等信息。
                .retryWhen(Retry.from(companion ->
                        // 在这里，我们将前三个错误视为可重试 （），然后放弃。take(3)
                        companion.take(3)));
        SubscribeUtils.subscribe(flux);
    }

    /**
     * Retry.backoff: 相比之下，Retry.backoff 是 Retry 对象的一部分，它表示一种特定的重试策略，即指数补偿（Exponential Backoff）。
     * 它会根据指定的初始延迟和最大延迟，在每次重试时以指数增长的方式延长重试间隔。
     * 使用标准回退策略（带抖动的指数回退）：（Flux|Mono）（另请参阅重试中的其他工厂方法retryWhen(Retry.backoff(...)))
     */
    @Test
    public void backoff() throws InterruptedException {
        Flux<String> flux = Flux
                // 这会不断产生错误，需要重试。
                .<String>error(new IllegalArgumentException())
                // doOnError在重试之前，我们可以记录并查看所有失败。
                .doOnError(e -> log.info("error : {}", e.toString()))
                // Retry.backoff: 相比之下，Retry.backoff 是 Retry 对象的一部分，它表示一种特定的重试策略，即指数补偿（Exponential Backoff）。
                // 它会根据指定的初始延迟和最大延迟，在每次重试时以指数增长的方式延长重试间隔。
                .retryWhen(Retry.backoff(3, Duration.ofMillis(1000))
                        // maxBackoff 方法允许您指定最大延迟。如果不指定，则默认为 30 秒。
                        .maxBackoff(Duration.ofMillis(3000)));
        SubscribeUtils.subscribe(flux);
        TimeUnit.SECONDS.sleep(5);
    }

    /**
     * 我想处理背压“错误”（从上游请求最大值，并在下游没有产生足够的请求时应用策略）......
     */

    /**
     * 通过抛出一个特殊的 IllegalStateException： Flux#onBackpressureError
     * <p>
     * Flux#onBackpressureError 是 Reactor Core 中 Flux 类的一个方法，用于定义在背压发生时的处理方式。
     * 在响应式流中，背压指的是当生产者生成事件速度大于消费者处理事件的速度时所引发的问题。
     * <p>
     * 具体来说，Flux#onBackpressureError 会在发生背压时抛出一个异常，通知生产者以减缓事件的生成速度或采取其他适当的措施。
     * 这样，如果消费者无法及时处理数据，生产者就会得到一个信号，表明需要减慢数据的生成速度，以平衡生产者和消费者之间的速率差异。
     */
    @Test
    public void onBackpressureError() {
        log.info("Flux#onBackpressureError 用于在背压发生时抛出异常，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。");
        /**
         * 在上面的例子中，如果生产者生成事件的速度超过了消费者处理事件的速度，onBackpressureError 会抛出异常，通知生产者限制事件的生成速度。
         * 总之，Flux#onBackpressureError 用于在背压发生时抛出异常，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
         */
        Flux.range(1, 10)
                // Flux#onBackpressureError 用于在背压发生时抛出异常，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
                .onBackpressureError().subscribe(new BaseSubscriber<>() {
                    @Override
                    protected void hookOnSubscribe(Subscription subscription) {
                        request(1);
                    }

                    @Override
                    protected void hookOnNext(Integer value) {
                        log.info("onNext : {}", value);
                        try {
                            TimeUnit.MILLISECONDS.sleep(100);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                });
    }

    /**
     * 通过删除多余的值：Flux#onBackpressureDrop
     * <p>
     * Flux#onBackpressureDrop 是 Reactor Core 中 Flux 类的一个方法，用于定义在背压发生时的处理方式。
     * 在响应式流中，背压指的是当生产者生成事件速度大于消费者处理事件的速度时所引发的问题。
     * <p>
     * 具体来说，Flux#onBackpressureDrop 会在发生背压时删除多余的值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
     */
    @Test
    public void onBackpressureDrop() {
        log.info("Flux#onBackpressureDrop 用于在背压发生时删除多余的值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。");
        /**
         * 在上面的例子中，如果生产者生成事件的速度超过了消费者处理事件的速度，onBackpressureDrop 会删除多余的值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
         * 总之，Flux#onBackpressureDrop 用于在背压发生时删除多余的值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
         */
        Flux.range(1, 10)
                // Flux#onBackpressureDrop 用于在背压发生时删除多余的值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
                .onBackpressureDrop().subscribe(new BaseSubscriber<>() {
                    @Override
                    protected void hookOnSubscribe(Subscription subscription) {
                        request(1);
                    }

                    @Override
                    protected void hookOnNext(Integer value) {
                        log.info("onNext : {}", value);
                        try {
                            TimeUnit.MILLISECONDS.sleep(100);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                });
    }

    /**
     * 除了最后一个：Flux#onBackpressureLatest
     * <p>
     * Flux#onBackpressureLatest 是 Reactor Core 中 Flux 类的一个方法，用于定义在背压发生时的处理方式。
     * 在响应式流中，背压指的是当生产者生成事件速度大于消费者处理事件的速度时所引发的问题。
     * <p>
     * 具体来说，Flux#onBackpressureLatest 会在发生背压时保留最后一个值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
     */
    @Test
    public void onBackpressureLatest() {
        log.info("Flux#onBackpressureLatest 用于在背压发生时保留最后一个值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。");
        /**
         * 在上面的例子中，如果生产者生成事件的速度超过了消费者处理事件的速度，onBackpressureLatest 会保留最后一个值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
         * 总之，Flux#onBackpressureLatest 用于在背压发生时保留最后一个值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
         */
        Flux.range(1, 10)
                // Flux#onBackpressureLatest 用于在背压发生时保留最后一个值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
                .onBackpressureLatest()
                .subscribe(new BaseSubscriber<>() {
                    @Override
                    protected void hookOnSubscribe(Subscription subscription) {
                        request(1);
                    }

                    @Override
                    protected void hookOnNext(Integer value) {
                        log.info("onNext : {}", value);
                        try {
                            TimeUnit.MILLISECONDS.sleep(100);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                });
    }

    /**
     * 通过缓冲多余的值（有界或无界）：Flux#onBackpressureBuffer
     * 并在有界缓冲区也溢出时应用策略：Flux#onBackpressureBuffer 和 BufferOverflowStrategy
     */
    @Test
    public void onBackpressureBuffer() {
        log.info("Flux#onBackpressureBuffer 用于在背压发生时缓冲多余的值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。");
        /**
         * 在上面的例子中，如果生产者生成事件的速度超过了消费者处理事件的速度，onBackpressureBuffer 会缓冲多余的值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
         * 总之，Flux#onBackpressureBuffer 用于在背压发生时缓冲多余的值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
         */
        Flux.range(1, 10)
                // Flux#onBackpressureBuffer 用于在背压发生时缓冲多余的值，以便进行适当的处理和调整事件流的速率，确保生产者和消费者之间的平衡。
                .onBackpressureBuffer()
                .subscribe(new BaseSubscriber<>() {
                    @Override
                    protected void hookOnSubscribe(Subscription subscription) {
                        request(1);
                    }

                    @Override
                    protected void hookOnNext(Integer value) {
                        log.info("onNext : {}", value);
                        try {
                            TimeUnit.MILLISECONDS.sleep(100);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                });
    }


}
